package com.hc.websocket.sse.service.impl;

import com.hc.websocket.sse.service.SseService;
import org.springframework.stereotype.Service;
import org.springframework.web.servlet.mvc.method.annotation.SseEmitter;

import java.io.IOException;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.function.Consumer;

@Service
public class SseServiceImpl implements SseService {
    private static final Map<String, SseEmitter> sseCache = new ConcurrentHashMap<>();
    /**
     * 创建一个新的Server-Sent Events连接。
     *
     * @param clientId 客户端的唯一标识符，用于管理和识别不同的客户端连接。
     * @return 返回一个SseEmitter实例，该实例可用于向客户端发送Server-Sent Events。
     */
    @Override
    public SseEmitter createConnect(String clientId){
        // 创建一个SseEmitter实例，设置其超时时间为0L，即不超时。
        SseEmitter sseEmitter = new SseEmitter(0L);

        // 为SseEmitter实例设置完成、超时和错误回调函数。
        sseEmitter.onCompletion(completionCallBack(clientId));
        sseEmitter.onTimeout(sendTimeoutCallBack(clientId));
        sseEmitter.onError(errorCallBack(clientId));

        // 将该SseEmitter实例与clientId关联，存储到sseCache中。
        sseCache.put(clientId, sseEmitter);

        // 打印连接创建信息。
        System.out.println("clientId:"+clientId+" connection created");

        try{
            // 尝试向客户端发送"hello"消息。
            // 客户端第一次连接必须发送消息，才能触发onopen事件，表示连接成功
            // 如果不发送，就不会触发onopen事件，后续发送消息才会触发
            sseEmitter.send("hello");
        }catch(IOException e){
            // 如果发送失败，打印错误信息。
            System.out.println("clientId:"+clientId+" send hello error");
        }

        return sseEmitter;
    }

    @Override
    public SseEmitter getSseEmitterByClientId(String clientId) {
        return sseCache.get(clientId);
    }

    @Override
    public void sendMessageToAllClient(String msg) {
        for (String key : sseCache.keySet()){
            try {
                sseCache.get(key).send(msg);
            } catch (IOException e) {
                System.out.println("clientId:"+key+" send error");
            }
        }
    }

    @Override
    public void sendMessageToOneClient(String clientId, String msg) {
        if (sseCache.containsKey(clientId)){
            try {
                sseCache.get(clientId).send(msg);
            } catch (IOException e) {
                System.out.println("clientId:"+clientId+" send error");
            }
        }

    }

    @Override
    public void closeConnect(String clientId) {
        if (sseCache.containsKey(clientId)){
            sseCache.get(clientId).complete();
            sseCache.remove(clientId);
        }
    }

    private Runnable completionCallBack(String clientId){
        return () -> {
            System.out.println("clientId:"+clientId+" connection closed");
            sseCache.remove(clientId);
        };
    }

    private Runnable sendTimeoutCallBack(String clientId){
        return () -> {
            System.out.println("clientId:"+clientId+" send timeout");
            sseCache.remove(clientId);
        };
    }

    private Consumer<Throwable> errorCallBack(String clientId) {
        return throwable -> {
            System.out.println("clientId:"+clientId+" error");
            sseCache.remove(clientId);
        };
    }
}